//Copyright © 2012 Ryan Larson, Ben Glasser
//[This program is licensed under the "GPL v3.0 License"]
// Please see the file COPYING in the source
//distribution of this software for license terms.

package org.app_chat.rabbitmq;

import com.google.common.base.Throwables;
import com.google.common.collect.Sets;
import com.google.protobuf.*;
import com.rabbitmq.client.*;
import org.app_chat.api.Listener;
import org.app_chat.api.Publisher;
import org.app_chat.service.protobuf.MessageUnMarshaller;

import java.io.IOException;
import java.util.Set;

/**
 * RabbitMQ implementation of the ChatRoom
 */

public class ChatRoom {
    private static final String EXCHANGE_NAME = "app-chat";

    private final String host;
    private final boolean durable;
    private final Set<Listener> subscriptions;

    private PublisherImpl publisher;

    private ChatRoom(Builder builder) {
        subscriptions = builder.subscriptions;
        durable = builder.durable;

        if (builder.host == null) {
            host = ConnectionFactory.DEFAULT_HOST;
        } else {
            host = builder.host;
        }
    }

    /**
     * Configures the queues for the listeners that have been
     * registered with {@link Builder#listenFor(org.app_chat.api.Listener)}
     * 
     * and creates the connection to the RabbitMQ Message Service
     */

    public void startChat() {

        try {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(host);

            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();

            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

            for (final Listener listener : subscriptions) {

                String queueName = listener.getQueueName();
                channel.queueDeclare(queueName, false, durable, false, null);
                channel.queueBind(queueName, EXCHANGE_NAME, "");

                channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope,
                                               AMQP.BasicProperties properties, byte[] messageBytes) {

                        try {
                            listener.messageReceived(MessageUnMarshaller.getUnMarshaller(listener).unMarshall(messageBytes));
                        } catch (Exception e) {
                            throw Throwables.propagate(e);
                        }
                    }
                });
            }
        } catch (IOException e) {
            throw Throwables.propagate(e);
        }

        publisher = new PublisherImpl();
    }

    /**
     * Closes all connections and shuts down the client connection
     * to the RabbitMQ message serivce
     */
    
    public void stopChat() {
        //TODO Implement this
    }

    /**
     * Gets the Publisher for this client instance. If a listener has not
     * already been registered, a queue will automatically be created
     * for the message with it's defauly queue name
     * 
     * @return  {@link Publisher} for this Client instance
     */

    public Publisher getPublisher() {
        return publisher;
    }

    /**
     * Static method to return the builder. This allows you to configure
     * the client conenction and add the listeners.
     * 
     * @return  {@link Builder} to construct a RabbitMQ client connection
     */

    public static Builder builder() {
        return new Builder();
    }

    /**
     * RabbitMQ Implementation of the {@link Publisher}
     */
    
    public class PublisherImpl implements Publisher {

        /**
         * RabbiMQ Publisher
         * 
         * @param message   The {@link Message} to be published
         */
        
        @Override
        public void publish(Message message) {
            String defaultQueueName = message.getDescriptorForType().getFullName();

            try {

                ConnectionFactory factory = new ConnectionFactory();
                factory.setHost(host);

                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel();

                channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

                channel.basicPublish(EXCHANGE_NAME, "", null, message.toByteArray());

            } catch (IOException e) {
                throw Throwables.propagate(e);
            }
        }
    }

    /**
     * A builder to construct the RabbitMQ ChatRoom
     */

    public static class Builder {

        private String host;
        private String userName;
        private String password;

        private boolean durable = false;

        Set<Listener> subscriptions = Sets.newHashSet();

        /**
         * Sets the host name. If this is {@literal null}, then
         * it will use the {@link ConnectionFactory#DEFAULT_HOST}
         * 
         * @param host  the hostname of the server where the RabbitMQ serive is running
         * @return  the {@link Builder} instance
         */
        
        public Builder setHost(String host) {
            this.host = host;
            return this;
        }

        /**
         * Sets the user name. If this is {@literal null}, then
         * it will not enter a userName/password
         * 
         * @param userName The userName required by the RabbitMQ server
         * @return  the {@link Builder} instance
         */

        public Builder setUserName(String userName) {
            this.userName = userName;
            return this;
        }

        /**
         * Sets the password.
         * 
         * @param password  The password required by the RabbitMQ server
         * @return the {@link Builder} instance
         */

        public Builder setPassword(String password) {
            this.password = password;
            return this;
        }

        /**
         * Sets the queue as durable, so the messages persist on disk
         * in the event of a Broker shutdown
         * 
         * @param durable   true for durable, false if not
         * @return  the {@link Builder} instance
         */

        public Builder setDurable(boolean durable) {
            this.durable = durable;
            return this;
        }

        public Builder listenFor(Listener listener) {
            this.subscriptions.add(listener);
            return this;
        }

        /**
         * Construct the ChatRoom instance
         * 
         * @return the {@link ChatRoom}
         */

        public ChatRoom build() {
            return new ChatRoom(this);
        }
    }
}
